package edu.zju.gis.dldsj.type;

/**
 * @author yanlo yanlong_lee@qq.com
 * @version 1.0 2018/07/17
 */

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

/**
 * where和group相关的查询尚未完成。
 */
public class StructFileQuerier {
    private Dataset<Row> data;
    private Map<String, Integer> fieldsIndex;
    private String select;
    private String where;
    private List<String> filter;
    private String orderby;
    private String groupby;
    private long offset;
    private long size;

    public StructFileQuerier(Dataset<Row> _sourceData, String _select, String _where,
                             List<String> _filter, String _orderby, String _groupby, long _offset, long _size) {
        data = _sourceData;
        fieldsIndex = data.getStructType().nameToIndex();
        select = _select;
        where = _where;
        filter = _filter;
        orderby = _orderby;
        groupby = _groupby;
        offset = _offset;
        size = _size;
    }

    public Dataset<Row> execute() {
        Stream<Row> stream = explainLimitConditions(data, offset, size, false);
        List<Row> MidResult = stream.filter(explainFilterConditions(filter)).sorted(explainOrderByConditions(orderby)).collect(Collectors.toList());
        return explainSelectConditions(data.getStructType(), MidResult, select);
    }

    /**
     * Comments: 解析参数limit传入的字段
     *
     * @param offset   查询忽略的行数
     * @param size     需要查询的行数
     * @param parallel 是否需要并行
     * @return 返回值为Stream<Row>
     */
    private static Stream<Row> explainLimitConditions(Dataset<Row> data, long offset, long size, boolean parallel) {
        if (size == 0)
            size = data.getCount();
        return StreamSupport.stream(data.spliterator(), parallel).skip(offset).limit(size);
    }

    /**
     * Comments: 解析参数select传入的字段
     *
     * @param sourceType 初始字段结构
     * @param sourceRows 初始值链表
     * @param conditions 查询字段，以逗号隔开
     * @return 返回值为Dataset<Row>
     */
    private Dataset<Row> explainSelectConditions(StructType sourceType, List<Row> sourceRows, String conditions) {
        StructType newType = sourceType;
        List<Row> FinalRows = null;
        if (select.isEmpty() || select.trim().equals("*")) {
            FinalRows = sourceRows;
        } else {
            FinalRows = new ArrayList<Row>();
            String[] ReqFields = select.split(",");
            int[] ReqFieldsIndex = new int[ReqFields.length];
            List<StructField> newFields = new ArrayList<StructField>();
            for (int i = 0; i < ReqFieldsIndex.length; i++) {
                ReqFieldsIndex[i] = fieldsIndex.get(ReqFields[i]);
                newFields.add(new StructField(ReqFields[i], DataType.STRING));
            }
            newType = new StructType(newFields);
            for (Row r : sourceRows) {
                String[] oneRow = new String[ReqFieldsIndex.length];
                for (int i = 0; i < ReqFieldsIndex.length; i++) {
                    oneRow[i] = r.getAsString(ReqFieldsIndex[i]);
                }
                FinalRows.add(new Row(oneRow));
            }
        }
        return new Dataset<Row>(newType, FinalRows);
    }

    /**
     * Comments: 解析orderby的字段
     *
     * @param conditions 排序字段，以逗号隔开
     * @return 返回值为Comparator
     */
    private Comparator<Row> explainOrderByConditions(String conditions) {
        if (conditions.isEmpty())
            return (o1, o2) -> 0;
        String[] orderFields = conditions.split(",");
        Comparator<Row> cmpList = Comparator.comparing(row -> row.getAsString(fieldsIndex.get(orderFields[0])));
        for (int i = 1; i < orderFields.length; i++) {
            int tmp = i;//final or effectively final
            cmpList = cmpList.thenComparing(row -> row.getAsString(fieldsIndex.get(orderFields[tmp])));
        }
        return cmpList;
    }

    /**
     * Comments: 解析filter的条件
     *
     * @param conditions 过滤条件，包含>=,<=,>,<,=
     * @return 返回值为Predicate<Row>
     */
    private Predicate<Row> explainFilterConditions(List<String> conditions) {
        Predicate<Row> preList = (n) -> true;
        for (String condition : conditions) {
            if (condition.contains(FilterConditions.GreaterOrEqual.getValue())) {
                String[] params = condition.split(">=");
                preList = preList.and(s -> s.getAsDouble(fieldsIndex.get(params[0].trim())) >= Double.parseDouble(params[1].trim()));
            } else if (condition.contains(FilterConditions.LessOrEqual.getValue())) {
                String[] params = condition.split("<=");
                preList = preList.and(s -> s.getAsDouble(fieldsIndex.get(params[0].trim())) <= Double.parseDouble(params[1].trim()));
            } else if (condition.contains(FilterConditions.Greater.getValue())) {
                String[] params = condition.split(">");
                preList = preList.and(n -> n.getAsDouble(fieldsIndex.get(params[0].trim())) > Double.parseDouble(params[1].trim()));
            } else if (condition.contains(FilterConditions.Less.getValue())) {
                String[] params = condition.split("<");
                preList = preList.and(n -> n.getAsDouble(fieldsIndex.get(params[0].trim())) < Double.parseDouble(params[1].trim()));
            } else if (condition.contains(FilterConditions.Equal.getValue())) {
                String[] params = condition.split("=");
                preList = preList.and(n -> n.get(fieldsIndex.get(params[0].trim())).equals(params[1].trim()));
            }
        }
        return preList;
    }

    private enum FilterConditions {
        Greater(">"),
        GreaterOrEqual(">="),
        Equal("="),
        LessOrEqual("<="),
        Less("<");
        String value;

        FilterConditions(String s) {
            this.value = s;
        }

        public String getValue() {
            return value;
        }
    }
}
